package nettyRPC.proxy;

import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import nettyRPC.protocol.RpcRequest;
import nettyRPC.protocol.RpcResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * @program: NettyRpc-master
 * @description
 * @author: tkk fendoukaoziji
 * @create: 2019-08-13 17:52
 **/
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {
    private static final Logger logger=LoggerFactory.getLogger(RpcClientHandler.class);
    private ConcurrentHashMap<String,RpcFuture> pendingRpc=new ConcurrentHashMap<>();

    private volatile Channel channel;
    private SocketAddress remotePeer;

    public Channel getChannel(){
        return channel;
    }
    public SocketAddress getRemotePeer(){
        return remotePeer;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.remotePeer=this.channel.remoteAddress();
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel=ctx.channel();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
        String requstId=response.getRequestId();
        RpcFuture rpcFuture = pendingRpc.get(requstId);
        if(rpcFuture!=null){
            pendingRpc.remove(requstId);
            rpcFuture.done(response);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("client caught exception", cause);
        ctx.close();
    }

    public RpcFuture sendRequest(RpcRequest request){
        final CountDownLatch latch=new CountDownLatch(1);
        RpcFuture rpcFuture = new RpcFuture(request);
        pendingRpc.put(request.getRequestId(),rpcFuture);
        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                latch.countDown();
            }
        });
        try {
            latch.await();
        } catch (InterruptedException e) {
            logger.error(e.getMessage());
        }
        return rpcFuture;
    }

    public void close(){
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }
}
















